package com.quanyan.mq.mq.consumer;

import com.google.common.base.Strings;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.mq.enums.MqConsumerStatus;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @Author Hugo.Wwg
 * @Since 2016-06-07
 */
public abstract class AbstractMqConsumer {

    private static final Logger log = LoggerFactory.getLogger(AbstractMqConsumer.class);

    private DefaultMQPushConsumer consumer;

    private String consumerGroup;

    private String nameServerAddress;

    private String instanceName;

    private MqTopic[] mqTopics;

    public void initConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(getConsumerGroup());//保证应用内唯一
        consumer.setNamesrvAddr(getNameServerAddress());
        consumer.setInstanceName(getInstanceName());//保证应用内唯一
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        mqTopics = getMqTopics();
        if (mqTopics != null && mqTopics.length > 0) {
            for (MqTopic mqTopic : mqTopics) {
                consumer.subscribe(mqTopic.getTopicName(),
                        Strings.isNullOrEmpty(mqTopic.getTags()) ? "*" : mqTopic.getTags());
            }
        }

    }

    public void start() {
        try {
            initConsumer();
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages,
                                                                ConsumeConcurrentlyContext context) {
                    //消费前根据全局唯一的msgId在Db中判断是否消费

                    /**
                     * 默认messages里只有一条消息，可以设置consumeMessageBatchMaxSize参数来批量接收消息
                     * consumer.setConsumeMessageBatchMaxSize(8);
                     **/

                    log.warn(Thread.currentThread().getName() + " receive messages: "
                            + messages.toString() + "size: " + messages.size());

                    MqConsumerStatus mqConsumerStatus = process(messages, context);

                    if (mqConsumerStatus.equals(MqConsumerStatus.RE_CONSUME_LATER)) {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumer.start();

        } catch (MQClientException e) {
            e.printStackTrace();
        }

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                consumer.shutdown();
                log.info("consumer调用addShutdownHook方法优雅关闭 ！！！");
            }
        }));
    }

    public void shutdown() {
        consumer.shutdown();
    }

    public void suspend() {
        consumer.suspend();
    }

    public void resume() {
        consumer.resume();
    }

    protected abstract MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context);

    protected abstract String getConsumerGroup();

    protected abstract String getNameServerAddress();

    protected abstract String getInstanceName();

    protected abstract MqTopic[] getMqTopics();
}
